package com.hmeter.hadoop.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * Component name: HBasePutSampler
 * Class used for HBase Delete operations that supports following functionalities:
 * 1. Create or update column qualifiers in HBase row (add or update list of column qualifiers in particular row on latest timestamp)
 * 
 * @author Bakir Jusufbegovic
 * @since 0.1
 * @version 0.1
 */
public class HBasePutSamplerUtility {
	
	private HBaseUtility hbaseUtility = null;
	
	public HBasePutSamplerUtility(String zooKeeperQuorum, String hbaseTable) throws IOException 
	{
		hbaseUtility = new HBaseUtility(zooKeeperQuorum, hbaseTable);
	}
	
	public HBaseUtility getHBaseUtility()
	{
		return hbaseUtility;
	}
	
	public String createOrUpdateColumnQualifiersInHBase(String rowKey, String fullColumnQualifiers, String filterColumnFamilies, String delimiter) throws IOException
	{	
		String response = "";
		List<Put> puts = new ArrayList<Put>();
		Put put = new Put(Bytes.toBytes(rowKey));
		
		if (delimiter.equals(""))
		{
			delimiter = delimiter.concat(",");
			fullColumnQualifiers = fullColumnQualifiers.concat(",");
			filterColumnFamilies = filterColumnFamilies.concat(",");
		}

		ArrayList<String> parsedList = hbaseUtility.parseElementsToList(fullColumnQualifiers, delimiter);
		long timestamp = hbaseUtility.getMostRecentTimestamp(rowKey, hbaseUtility.parseElementsToList(filterColumnFamilies, delimiter));

		String[] columnFamilyColumnValueToken = null;
		String[] columnValueToken = null;

		try {
			for (int i=0; i<parsedList.size(); i++) {
				columnFamilyColumnValueToken = parsedList.get(i).split(":");
				columnValueToken = columnFamilyColumnValueToken[1].split("=");
				put.add(Bytes.toBytes(columnFamilyColumnValueToken[0]), Bytes.toBytes(columnValueToken[0]), Long.valueOf(timestamp), Bytes.toBytes(columnValueToken[1]));
				puts.add(put);
			}
			hbaseUtility.getHTable().put(puts);
			response = "Added/Updated list of column qualifiers: " + fullColumnQualifiers;
		}
		catch (IOException ex) {
			response = ex.getMessage();
		}
		
		return response;
	}	
}
